package tom.vertx.core.demo02;

import io.vertx.core.AbstractVerticle;
import io.vertx.core.Promise;
import io.vertx.core.eventbus.DeliveryOptions;
import io.vertx.core.eventbus.impl.codecs.StringMessageCodec;
import io.vertx.core.json.JsonObject;
import lombok.extern.slf4j.Slf4j;

import java.time.LocalDateTime;

/**
 * @author ZHUFEIFEI
 */
@Slf4j
public class MyVerticle02 extends AbstractVerticle {
  @Override
  public void start(Promise<Void> startPromise) throws Exception {

    //注册class-codec对应关系的codec
//    vertx.eventBus().registerDefaultCodec(clazz, codec);
    vertx.eventBus().registerCodec(new StringMessageCodec() {
      @Override
      public String name() {
        return "tom.codec.server";
      }
      @Override
      public byte systemCodecID() {
        return -1;
      }
    });

    vertx.setPeriodic(15000L, id -> {
      log.info("publish message.");

      //pub/sub
      vertx.eventBus()
        .publish("tom.msg.bus", "hello event bus.", new DeliveryOptions().addHeader("a", "b"));

      //p2p
      vertx.eventBus().send("tom.p2p", "hello event bus p2p.");
      vertx.eventBus().send("tom.codec.demo", "hello event bus codec."
        , new DeliveryOptions().setCodecName("tom.codec.server"));
      vertx.eventBus().send("tom.p2p.json", new JsonObject().put("name","tom"));

      //request-response, 一直reply失败 ,如果只留这一个请求则正常, 由于上面设置了codec,导致消息无法被正常接收解析处理, codec写的不对, 需要重写encodeToWire/decodeFromWire
      //因为codec最终操作会将数据按照类型写入buffer, 接收端buffer获取字段数据长度,然后在从pos位置读取长度个字节转为字符串. 对于自定义codec, 每个verticle都要注册才能正常解析发送数据
      vertx.eventBus().request("tom.request", "request something", new DeliveryOptions().setSendTimeout(1500L), reply -> {
        if (reply.succeeded()) {
          log.info("reply => {}", reply.result().body());
        } else {
          log.info("relay failed => {}", reply.cause().toString());
        }
      });


      //发到不存在的地址
      vertx.eventBus().request("tom.not.exist.consumer", "request null",new DeliveryOptions().setSendTimeout(10L), reply -> {
        if (reply.succeeded()) {
          log.info("reply => {}", reply.result().body());
        } else {
          //No handlers for address
          log.info("send failed => {}", reply.cause().toString());
        }
      });

    });

    startPromise.complete();
  }
}
